package com.ray.dfj;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.SequenceGroup;
import com.lmax.disruptor.SingleThreadedClaimStrategy;

/**
 * The Class Generator.
 *
 * @param <T> The event type
 */
public class Generator<T> {
	private Task task;
	private RingBuffer<T> ringBuffer;
	private int bufferSize = 64;

	private List<EventProcessor> eventProcessorList = new LinkedList<EventProcessor>();
	private EventFactory<T> eventFactory;
	private BlockingWaitStrategy waitStrategy = new BlockingWaitStrategy();

	/**
	 * Gets the task.
	 *
	 * @return the task
	 */
	public Task getTask() {
		return task;
	}

	/**
	 * Sets the task.
	 *
	 * @param task the new task
	 */
	public void setTask(Task task) {
		this.task = task;
	}

	/**
	 * Start taking requests for the ring buffer.
	 */
	public void start() {
		// submit all processors
		ExecutorService threadPool = Executors.newFixedThreadPool(eventProcessorList.size());
		for (EventProcessor item : eventProcessorList) {
			threadPool.submit(item);
		}
	}

	/**
	 * Initiate the generator by constructing the sequence barriers from the task model.
	 */
	public void init() {

		if (eventFactory == null)
			throw new NullPointerException("EventFactory is not set");
		ringBuffer = new RingBuffer<T>(eventFactory, new SingleThreadedClaimStrategy(bufferSize), waitStrategy);
		final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

		Sequence lastSequence = processTask(this.task, sequenceBarrier);
		ringBuffer.setGatingSequences(lastSequence);

	}

	@SuppressWarnings("unchecked")
	private Sequence processTask(Task task, SequenceBarrier sequenceBarrier) {
		if (task instanceof MultiTask) {
			// Handle empty list
			MultiTask mTask = (MultiTask) task;
			if (mTask.getTasks().isEmpty()) {
				return null;
			}

			if (task instanceof ParallelTask) {
				ParallelTask pTask = (ParallelTask) task;
				SequenceGroup sequenceGroup = new SequenceGroup();
				for (Task item : pTask.getTasks()) {
					Sequence pSeq = processTask(item, sequenceBarrier);
					if (pSeq != null)
						sequenceGroup.add(pSeq);
				}
				return sequenceGroup;
			} else if (task instanceof SerialTask) {
				SerialTask sTask = (SerialTask) task;
				SequenceBarrier nextBarrier = sequenceBarrier;
				Sequence lastSequence = null;
				for (Task item : sTask.getTasks()) {
					Sequence sSeq = processTask(item, nextBarrier);
					if (sSeq == null)
						continue;
					lastSequence = sSeq;
					nextBarrier = ringBuffer.newBarrier(lastSequence);
				}
				return lastSequence;
			}
		} else if (task instanceof SingleTask) {
			SingleTask<T> sTask = (SingleTask<T>) task;
			BatchEventProcessor<T> eventProcessor = new BatchEventProcessor<T>(ringBuffer, sequenceBarrier, (EventHandler<T>) task);
			if (sTask.getExceptionHandler() != null) {
				eventProcessor.setExceptionHandler(sTask.getExceptionHandler());
			}
			eventProcessorList.add(eventProcessor);
			return eventProcessor.getSequence();
		}
		return null;
	}

	/**
	 * Gets the ring buffer.
	 *
	 * @return the ring buffer
	 */
	public RingBuffer<T> getRingBuffer() {
		return ringBuffer;
	}

	/**
	 * Gets the buffer size.
	 *
	 * @return the buffer size
	 */
	public int getBufferSize() {
		return bufferSize;
	}

	/**
	 * Sets the buffer size.
	 *
	 * @param bufferSize the new buffer size
	 */
	public void setBufferSize(int bufferSize) {
		this.bufferSize = bufferSize;
	}

	/**
	 * Gets the event factory.
	 *
	 * @return the event factory
	 */
	public EventFactory<T> getEventFactory() {
		return eventFactory;
	}

	/**
	 * Sets the event factory.
	 *
	 * @param eventFactory the new event factory
	 */
	public void setEventFactory(EventFactory<T> eventFactory) {
		this.eventFactory = eventFactory;
	}

}
